package cn.dmp.report

import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ListBuffer

/**
  * Created by Irving on 2018/3/2.
  */
object PCount {

    def main(args: Array[String]): Unit = {

        val conf = new SparkConf()
          .setMaster("local[*]")
          .setAppName(s"${this.getClass.getSimpleName}")
          .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

        val sc = new SparkContext(conf)
        val sqlc = new SQLContext(sc)

        val json = sqlc.read.parquet("F:\\Temp\\parquet1")

        val database = json.map(row => {

//            val list = new ListBuffer[Double]()
//
//            val provincename = row.getAs[String]("provincename")
//            val cityname = row.getAs[String]("cityname")
//            // 数据求情方式
//            val requestmode = row.getAs[Int]("requestmode")
//            // 流程节点
//            val processnode = row.getAs[Int]("processnode")
//            // 流程节点
//            val iseffective = row.getAs[Int]("iseffective")
//            // 是否收费
//            val isbilling = row.getAs[Int]("isbilling")
//            // 是否rtb
//            val isbid = row.getAs[Int]("isbid")
//            // 是否竞价成功
//            val iswin = row.getAs[Int]("iswin")
//            // 广告id
//            val adorderid = row.getAs[Int]("adorderid")
//            // 广告消费
//            val winprice = row.getAs[Double]("winprice")
//            // 广告成本
//            val adpayment = row.getAs[Double]("adpayment")
//            if (requestmode == 1 && processnode >= 1) list.append(1) else list.append(0)
//            if (requestmode == 1 && processnode >= 2) list.append(1) else list.append(0)
//            if (requestmode == 1 && processnode == 3) list.append(1) else list.append(0)
//            if (iseffective == 1 && isbilling == 1 && isbid == 1 && adorderid != 0) list.append(1) else list.append(0)
//            if (iseffective == 1 && isbilling == 1 && iswin == 1) list.append(1) else list.append(0)
//            if (requestmode == 2 && iseffective == 1) list.append(1) else list.append(0)
//            if (requestmode == 3 && iseffective == 1) list.append(1) else list.append(0)
//            if (iseffective == 1 && isbilling == 1 && iswin == 1) list.append(winprice / 1000) else list.append(0)
//            if (iseffective == 1 && isbilling == 1 && iswin == 1) list.append(adpayment / 1000) else list.append(0)

            val list = new ListBuffer[Double]()

            val pName = row.getAs[String]("provincename")
            val cName = row.getAs[String]("cityname")
            val rMode = row.getAs[Int]("requestmode")
            val pNode = row.getAs[Int]("processnode")
            val effect = row.getAs[Int]("iseffective")
            val bill = row.getAs[Int]("isbilling")
            val isbid = row.getAs[Int]("isbid")
            val iswin = row.getAs[Int]("iswin")
            val adorderid = row.getAs[Int]("adorderid")
            val winprice = row.getAs[Double]("winprice")
            val adpayment = row.getAs[Double]("adpayment")

            // 广告请求
            if (rMode == 1) {
                if (pNode == 1) {
                    list.append(1, 0, 0)
                } else if (pNode == 2) {
                    list.append(1, 1, 0)
                } else if (pNode == 3) {
                    list.append(1, 1, 1)
                } else list.append(0, 0, 0)
            } else {
                list.append(0, 0, 0)
            }

            // 广告参与竞价
            if (effect == 1 && bill == 1) {
                if (isbid == 1 && adorderid != 0) list.append(1, 0) else list.append(0,0)
                if (iswin == 1) list.append(0, 1) else list.append(0, 0)
            } else list.append(0, 0) // 参与竞价，竞价成功

            // 展示点击
            if (effect == 1) {
                if (rMode == 2) {
                    list.append(1, 0)
                } else if (rMode == 3) {
                    list.append(0, 1)
                } else list.append(0, 0)
            } else list.append(0, 0)

            // 消费成本
            if (effect == 1 && bill == 1 && iswin == 1) {
                list.append(winprice / 1000, adpayment / 1000)
            } else list.append(0, 0)

            // List(原始请求数,有效请求数,广告请求数,参与竞价数,竞价成功数,展示数,点击数,DSP广告消费，DSP广告成本)
            ((pName, cName), list)
        })


        // 地域分布报表
        database.reduceByKey{
            case (list1, list2) => list1.zip(list2).map(t => t._1 + t._2)
        }.map(t => t._1._1 + "," +t._1._2 + "," + t._2.mkString(","))
          .foreach(println)


        sc.stop()
    }
}
